%%% -------------------------------------------------------------------
%%% Author  : Administrator
%%% Description :
%%%
%%% Created : 2009-3-23
%%% -------------------------------------------------------------------
-module(node_readers).

-behaviour(gen_server).
%% --------------------------------------------------------------------
%% Include files
%% --------------------------------------------------------------------
-include("ring.hrl").
-include("map.hrl").

%% --------------------------------------------------------------------
%% External exports
-export([start_link/1]).
-export([get_current_reader/1, update_nodereaders/1, report_state/0]).

%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).

-record(node_readers, {
    node_id :: node_id(),
    readers = [] :: [pid()]
}). 

-record(state, {current_idx = 1}).

%% ====================================================================
%% External functions
%% ====================================================================
start_link(RingNodes) ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, RingNodes, []).


%% ====================================================================
%% Server functions
%% ====================================================================
get_current_reader(RingNodeId) ->
    gen_server:call(?MODULE, {get_current_reader_for, RingNodeId}).

update_nodereaders(RingNodes) ->
    gen_server:call(?MODULE, {update_nodereaders, RingNodes}).

report_state() ->
    gen_server:call(?MODULE, {report_state}).

%% --------------------------------------------------------------------
%% Function: init/1
%% Description: Initiates the server
%% Returns: {ok, State}          |
%%          {ok, State, Timeout} |
%%          ignore               |
%%          {stop, Reason}
%% --------------------------------------------------------------------
init(RingNodes) ->
    process_flag(trap_exit, true),

    ets:new(node_readers, [named_table, ordered_set, protected, {keypos, #node_readers.node_id}]),
    ets:new(reader_to_node, [named_table, ordered_set, protected]),

    new_nodereader(RingNodes),

    {ok, #state{}}.

%% --------------------------------------------------------------------
%% Function: handle_call/3
%% Description: Handling call messages
%% Returns: {reply, Reply, State}          |
%%          {reply, Reply, State, Timeout} |
%%          {noreply, State}               |
%%          {noreply, State, Timeout}      |
%%          {stop, Reason, Reply, State}   | (terminate/2 is called)
%%          {stop, Reason, State}            (terminate/2 is called)
%% --------------------------------------------------------------------
handle_call({report_state}, _From, #state{current_idx=Idx}=State) ->
    io:format("node_readers: ~n", []),
    lists:foreach(fun(#node_readers{node_id=NodeId, readers=ReaderPids}) ->
        io:format("node_reader_for: ~p, idx: ~p~n", [NodeId, Idx]),
        lists:foreach(fun(Pid) ->
            io:format("\t~p(~p)", [Pid, erlang:is_process_alive(Pid)])
        end, ReaderPids),
        io:format("~n")
    end, ets:tab2list(node_readers)),
    {reply, ok, State};
handle_call({update_nodereaders, RingNodes}, _From, State) ->
    terminate_nodereaders(),
    new_nodereader(RingNodes),
    {reply, ok, State};
handle_call({get_current_reader_for, RingNodeId}, _From, #state{current_idx=Idx}=State) ->
    [#node_readers{readers=ReaderPids}] = ets:lookup(node_readers, RingNodeId),
    NewIdx = (Idx rem length(ReaderPids)) + 1,
    Reader = lists:nth(NewIdx, ReaderPids),
%    ets:insert(node_readers, NodeReaders#node_readers{current_idx=NewIdx}),
    {reply, Reader, State#state{current_idx=NewIdx}}.

%% --------------------------------------------------------------------
%% Function: handle_cast/2
%% Description: Handling cast messages
%% Returns: {noreply, State}          |
%%          {noreply, State, Timeout} |
%%          {stop, Reason, State}            (terminate/2 is called)
%% --------------------------------------------------------------------
handle_cast(_Msg, State) ->
    {noreply, State}.

%% --------------------------------------------------------------------
%% Function: handle_info/2
%% Description: Handling all non call/cast messages
%% Returns: {noreply, State}          |
%%          {noreply, State, Timeout} |
%%          {stop, Reason, State}            (terminate/2 is called)
%% --------------------------------------------------------------------
handle_info({'EXIT', ReaderPid, Reason}, State) ->
    log4erl:debug("some writer(~p) exit with: ~p ", [ReaderPid, Reason]),
    case ets:lookup(reader_to_node, ReaderPid) of
        [{ReaderPid, RingNodeId}] -> 
            log4erl:info("Writer(~p) for node(~p) exit with: ~p~n", [ReaderPid, RingNodeId, Reason]),
            [#node_readers{readers=ReaderPids}=NodeReaders] = ets:lookup(node_readers, RingNodeId),
            NewReaderPids = lists:dropwhile(fun(Pid) -> Pid == ReaderPid end, ReaderPids),
            ets:delete(reader_to_node, ReaderPid),
            ets:insert(node_readers, NodeReaders#node_readers{readers=NewReaderPids}),
            {noreply, #state{current_idx=1}=State};
        [] ->
            {noreply, State}
    end;
handle_info(Info, State) ->
    log4erl:warn("some reader conn may break: ~p", [Info]),
    {noreply, State}.

%% --------------------------------------------------------------------
%% Function: terminate/2
%% Description: Shutdown the server
%% Returns: any (ignored by gen_server)
%% --------------------------------------------------------------------
terminate(_Reason, _State) ->
    terminate_nodereaders(),
    ok.

terminate_nodereaders() ->
    NodereaderList = ets:tab2list(node_readers),
    lists:foreach(fun(#node_readers{readers = OldNodeReaders}) -> 
        lists:foreach(fun(OldNodeReader) ->
            case erlang:is_process_alive(OldNodeReader) of
                true ->
                    OldNodeReader ! {stop};
                false ->
                    stopped
                end                            
            end, OldNodeReaders)
    end, NodereaderList), 
    ets:delete_all_objects(reader_to_node),
    ets:delete_all_objects(node_readers).

%% --------------------------------------------------------------------
%% Func: code_change/3
%% Purpose: Convert process state when code is changed
%% Returns: {ok, NewState}
%% --------------------------------------------------------------------
code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%% --------------------------------------------------------------------
%%% Internal functions
%% --------------------------------------------------------------------
new_nodereader(RingNodes) ->
    NodeReaders = lists:map(fun create_readers_for/1, RingNodes),
    lists:foreach(fun({NodeId, ReaderPids}) ->
        ets:insert(node_readers , #node_readers{node_id=NodeId, readers=ReaderPids}),
        lists:foreach(fun(ReaderPid) ->
            ets:insert(reader_to_node, {ReaderPid, NodeId})
        end, ReaderPids)
    end, NodeReaders).

create_readers_for(#ring_node{id=NodeId, ip=Ip, access_port=Port}) ->
    Readers = [new_reader(Ip, Port) || _ <- lists:seq(1, ?READER_POOL_SIZE)],
    {NodeId, Readers}.

new_reader(Address, Port) ->
    Pid = spawn_link(fun() ->
        {ok, Socket} = gen_tcp:connect(Address, Port, [binary, {packet, 4}, {active,once}]),
        read_loop(Socket)
    end),
    Pid.

read_loop(Socket) ->
    receive
        {get, Key, From} = _Msg ->
            gen_tcp:send(Socket, term_to_binary({get, Key})),
            receive
                {tcp, Socket, Bin} ->
                    Cell = binary_to_term(Bin),
                    %From ! Cell,
                    gen_server:reply(From, Cell),
                    inet:setopts(Socket, [{active, once}]),
                    read_loop(Socket);
                {tcp_closed, Socket} ->
                    closed;
                {tcp_error, Socket, Reason} ->
                    exit(Reason)
            end;
        {stop} ->
            stop
    end.
